package com.dm.server.utils;

import com.alibaba.google.common.util.concurrent.ThreadFactoryBuilder;
import com.dm.server.entity.VoucherOrder;
import com.dm.server.service.SeckillVoucherService;
import com.dm.server.service.VoucherOrderService;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import static cn.dev33.satoken.SaManager.log;

@Component
public class StreamOrderProcessor {

    
    @Resource
    private StringRedisTemplate stringRedisTemplate;
    
    @Resource
    private VoucherOrderService voucherOrderService;
    
    @Resource
    private SeckillVoucherService seckillVoucherService;
    
    @Resource
    private RedisIdWorker redisIdWorker;
    
    private static final String STREAM_KEY = "seckill:orders";
    private static final String CONSUMER_GROUP = "order-group";
    private static final String CONSUMER_NAME = "consumer-" + UUID.randomUUID().toString();
    
    // 核心线程池：专门处理Stream订单
    private final ThreadPoolExecutor orderExecutor = new ThreadPoolExecutor(
        20,                    // 核心线程数
        50,                    // 最大线程数
        60L,                   // 空闲线程存活时间
        TimeUnit.SECONDS,      // 时间单位
        new LinkedBlockingQueue<>(1000),  // 工作队列
        new ThreadFactoryBuilder().setNameFormat("stream-order-%d").build(),  // 线程工厂
        new ThreadPoolExecutor.CallerRunsPolicy()  // 拒绝策略
    );

    @PostConstruct
    public void init() {
        try {
            // 创建消费者组
            stringRedisTemplate.opsForStream().createGroup(STREAM_KEY, CONSUMER_GROUP);
            log.info("创建消费者组成功: {}", CONSUMER_GROUP);
        } catch (Exception e) {
            log.info("消费者组已存在: {}", CONSUMER_GROUP);
        }
        
        // 启动Stream监听线程
        startStreamListener();
    }
    
    /**
     * 启动Stream监听线程
     */
    private void startStreamListener() {
        Thread listenerThread = new Thread(() -> {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    // 从Stream读取消息
                    List <MapRecord<String, Object, Object>> messages = readFromStream();
                    
                    if (messages != null && !messages.isEmpty()) {
                        // 将消息提交到线程池处理
                        submitMessagesToThreadPool(messages);
                    }
                } catch (Exception e) {
                    log.error("Stream监听异常", e);
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                        break;
                    }
                }
            }
        }, "stream-listener");
        
        listenerThread.setDaemon(true);
        listenerThread.start();
        log.info("Stream监听线程启动成功");
    }
    
    /**
     * 从Stream读取消息
     */
    private List<MapRecord<String, Object, Object>> readFromStream() {
        //从哪个消费者读取，一次读取50条，没有消息阻塞等待50s，会从上次消费的位置读取
        return stringRedisTemplate.opsForStream().read(
            Consumer.from(CONSUMER_GROUP, CONSUMER_NAME),
            StreamReadOptions.empty().count(50).block(Duration.ofMillis(100)),
            StreamOffset.create(STREAM_KEY, ReadOffset.lastConsumed())
        );
    }
    
    /**
     * 将消息提交到线程池处理
     */
    private void submitMessagesToThreadPool(List<MapRecord<String, Object, Object>> messages) {
        for (MapRecord<String, Object, Object> message : messages) {
            // 提交到线程池处理
            orderExecutor.submit(() -> processOrderMessage(message));
        }
        log.info("提交{}条消息到线程池", messages.size());
    }
    
    /**
     * 处理订单消息（在线程池中执行）
     */
    private void processOrderMessage(MapRecord<String, Object, Object> message) {
        String messageId = String.valueOf(message.getId());
        try {
            log.info("线程{}开始处理消息: {}", Thread.currentThread().getName(), messageId);
            
            // 1. 解析消息数据
            Map<Object, Object> value = message.getValue();
            Long userId = Long.valueOf(value.get("userId").toString());
            Long voucherId = Long.valueOf(value.get("voucherId").toString());
            long orderId = redisIdWorker.nextId("order");

            // 2. 创建订单并扣减库存
            boolean success = createOrderAndDeductStock(userId, voucherId,orderId);
            
            if (success) {
                // 3. 确认消息
                log.info("消息{}处理成功", messageId);
            } else {
                log.error("消息{}处理失败", messageId);
            }
            
        } catch (Exception e) {
            log.error("处理消息{}异常", messageId, e);
        }
    }


    /**
     * 创建订单并扣减库存（事务操作）
     */
    private boolean createOrderAndDeductStock(Long userId, Long voucherId,Long orderId) {
        try {
            VoucherOrder voucherOrder = new VoucherOrder();
            voucherOrder.setId(orderId);
            voucherOrder.setUserId(userId);
            voucherOrder.setVoucherId(voucherId);
            voucherOrder.setCreateTime(LocalDateTime.now());
            // 使用事务确保订单创建和库存扣减的原子性
            return voucherOrderService.executeOrderTransaction(voucherOrder);
        } catch (Exception e) {
            log.error("创建订单失败 - 用户ID: {}, 商品ID: {}", userId, voucherId, e);
            return false;
        }
    }

    /**
     * 优雅关闭
     */
    @PreDestroy
    public void shutdown() {
        log.info("开始关闭Stream订单处理线程池...");
        orderExecutor.shutdown();
        try {
            if (!orderExecutor.awaitTermination(30, TimeUnit.SECONDS)) {
                orderExecutor.shutdownNow();
            }
        } catch (InterruptedException e) {
            orderExecutor.shutdownNow();
            Thread.currentThread().interrupt();
        }
        log.info("Stream订单处理线程池已关闭");
    }
}